package com.chinamsp.zookeeper;

import org.apache.curator.retry.BoundedExponentialBackoffRetry;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import org.springframework.integration.zookeeper.config.CuratorFrameworkFactoryBean;
import org.springframework.integration.zookeeper.lock.ZookeeperLockRegistry;
import org.springframework.integration.zookeeper.lock.ZookeeperLockRegistry.KeyToPathStrategy;
import org.springframework.integration.zookeeper.metadata.ZookeeperMetadataStore;
import org.springframework.util.Assert;

import com.chinamsp.baisc.util.ApplicationConstants;

/**
 * zookeeper 配置
 * @author shifenghu
 * @version 1.0
 */
@Configuration
public class ZookeeperFactoryBean implements DisposableBean{
	/**
	 * zookeeper 配置前缀
	 */
	public static final String PROPERTIES_PREFIX = ApplicationConstants.PROPERTIES_PREFIX +  ".zookeeper";
	
	private CuratorFrameworkFactoryBean zookeeperFactory;
	
	@Autowired
	private ZookeeperProperties properties;
	
	/**
	 * 创建一个 zookeeper factory bean
	 * @return
	 */
	@Bean
	@Order(Ordered.LOWEST_PRECEDENCE - 1)
	public CuratorFrameworkFactoryBean zkFactory(){
		Assert.notNull(properties.getAddress(), String.format("%s.%s 未配置", PROPERTIES_PREFIX, "address"));
		zookeeperFactory = new CuratorFrameworkFactoryBean(properties.getAddress(), new BoundedExponentialBackoffRetry(properties.getBaseSleepTimeMs(), properties.getMaxSleepTimeMs(), properties.getMaxRetries()));		
		//如果是自动启动，则开启  zookeeper 连接
		zookeeperFactory.setAutoStartup(properties.isAutoStart());
		return zookeeperFactory;
	}
	
	/**
	 * 是否开启 zookeeper
	 * @return
	 */
	public boolean isEnabled(){
		return properties.isAutoStart();
	}
	
	/**
	 * 创建一个 zk store
	 * @param factory
	 * @return
	 */
	public ZookeeperMetadataStore newStore(){
		try {
			/**
			 * 这里可以配置 权限
			 * List<ACL> acls = new ArrayList<ACL>(2);
			 * Id id1 = new Id("digest", DigestAuthenticationProvider.generateDigest("admin:admin123"));  
				ACL acl1 = new ACL(ZooDefs.Perms.ALL, id1);  
				  
				Id id2 = new Id("digest", DigestAuthenticationProvider.generateDigest("guest:guest123"));  
				ACL acl2 = new ACL(ZooDefs.Perms.READ, id2);  
				  
				acls.add(acl1);  
				acls.add(acl2);  
				zookeeperFactory.getObject().setACL().withACL(acls);
			 */
			return new ZookeeperMetadataStore(zookeeperFactory.getObject());
		} catch (Exception e) {
			throw new IllegalArgumentException(e);
		}
	}
	
	/**
	 * 初始化一个 lock
	 * @return
	 */
	public ZookeeperLockRegistry newLockRegistry(){
		try {
			return new ZookeeperLockRegistry(zookeeperFactory.getObject());
		} catch (Exception e) {
			throw new IllegalArgumentException(e);
		}
	}
	
	/**
	 * 初始化一个 lock
	 * @param root 锁的  path
	 * @return
	 */
	public ZookeeperLockRegistry newLockRegistry(String root){
		try {
			return new ZookeeperLockRegistry(zookeeperFactory.getObject(), root);
		} catch (Exception e) {
			throw new IllegalArgumentException(e);
		}
	}
	
	/**
	 * 初始化一个 lock
	 * @param path 锁的  path
	 * @return
	 */
	public ZookeeperLockRegistry newLockRegistry(KeyToPathStrategy path){
		try {
			return new ZookeeperLockRegistry(zookeeperFactory.getObject(), path);
		} catch (Exception e) {
			throw new IllegalArgumentException(e);
		}
	}

	@Override
	public void destroy() throws Exception {
		// 如果是运行状态，则停止
		if (zookeeperFactory.isRunning()){
			zookeeperFactory.stop();
		}
	}
	
	
}
